package testview.zhen.com.myapplication.design.consumer_producer;


import java.nio.Buffer;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 优化版本
 * 对于LockConditionModel进行优化，突破只能一个线程进行处理链表数据。
 * 两个Lock 两个Condition
 * 那么思路就简单了：需要两个锁 CONSUME_LOCK与PRODUCE_LOCK，CONSUME_LOCK控制消费者线程并发出队
 * ，PRODUCE_LOCK控制生产者线程并发入队；相应需要两个条件变量NOT_EMPTY与NOT_FULL，NOT_EMPTY负责控制消费者线程的状态（阻塞、运行），
 * NOT_FULL负责控制生产者线程的状态（阻塞、运行）。
 * 以此让优化消费者与消费者（或生产者与生产者）之间是串行的；
 * 消费者与生产者之间是并行的。
 */
public class LockConditionModel2 implements Model {

    private final Lock CONSUME_LOCK = new ReentrantLock();
    private final Condition NOT_EMPTY = CONSUME_LOCK.newCondition();

    private final Lock PRODUCE_LOCK = new ReentrantLock();
    private final Condition NOT_FULL = PRODUCE_LOCK.newCondition();

    private final Buffer<Task> buffer = new Buffer<>();
    private AtomicInteger bufLen = new AtomicInteger(0);
    private final AtomicInteger increTaskNo = new AtomicInteger(0);
    private final int cap;

    public LockConditionModel2(int cap){
        this.cap = cap;
    }

    @Override
    public Runnable newRunnableConsumer() {
        return new ConsumerImpl();
    }

    @Override
    public Runnable newRunnableProducer() {
        return new ProducerImpl();
    }

    private class ConsumerImpl extends AbstractConsumer implements Consumer,Runnable{

        @Override
        public void consume() throws InterruptedException {
            int newBufSize = -1;
            CONSUME_LOCK.lockInterruptibly();
            try {
                while (bufLen.get() == 0){
                    System.out.println("buffer is empty...");
                    NOT_EMPTY.await();
                }
                Task task = buffer.poll();
                newBufSize = bufLen.decrementAndGet();
                assert task != null;
                // 固定时间范围的消费，模拟相对稳定的服务器处理过程
                Thread.sleep(500 + (long) (Math.random() * 500));
                System.out.println("consume: " + task.no);
                if (newBufSize > 0) {
                    NOT_EMPTY.signalAll();
                }
            }finally {
                CONSUME_LOCK.unlock();
            }
            if (newBufSize < cap){
                PRODUCE_LOCK.lockInterruptibly();
                try {
                    NOT_FULL.signalAll();
                }finally {
                    PRODUCE_LOCK.unlock();
                }
            }

        }
    }

    private class ProducerImpl extends AbstractProducer implements Producer,Runnable{

        @Override
        public void proudce() throws InterruptedException {

            // 不定期生产，模拟随机的用户请求
            Thread.sleep((long) (Math.random() * 1000));

            int newBufSize = -1;

            PRODUCE_LOCK.lockInterruptibly();
            try{
                while (bufLen.get() == cap){
                    System.out.println("buffer is full...");
                    NOT_FULL.await();
                }
                Task task = new Task(increTaskNo.getAndIncrement());
                buffer.offer(task);
                newBufSize = bufLen.incrementAndGet();
                System.out.println("produce: " + task.no);
                if (newBufSize < cap) {
                    NOT_FULL.signalAll();
                }
            }finally {
                PRODUCE_LOCK.unlock();
            }
            if (newBufSize > 0){
                CONSUME_LOCK.lockInterruptibly();
                try {
                    NOT_EMPTY.signalAll();
                }finally {
                    CONSUME_LOCK.unlock();
                }
            }

        }
    }

    /**
     * 在头部出队，尾部入队
     * 在poll()方法中只操作head
     * 在offer(E e)方法中只操作tail
     * @param <E>
     */
    public static class Buffer<E>{

        private Node head;
        private Node tail;
        private AtomicInteger count = new AtomicInteger(0) ;

        public Buffer(){
            head = tail = new Node(null);
        }

        public int getCount(){
            return count.get();
        }

        /**
         * 将数据添加到最后的tail.next上 然后将tail.next设置是为tail
         * @param e
         */
        public void offer(E e){
            head.hashCode();
            tail.next = new Node(e);
            tail = tail.next;
            int a = count.incrementAndGet();
            System.out.println("链表的长度为："+a);
        }

        /**
         * 获取head 的next赋予head 并且返回新的head.item  然后将head置空
         * @return
         */
        public E poll(){
            
            head = head.next;
            E e = head.item;
            head.item = null;
            int a = count.decrementAndGet();
            System.out.println("链表的长度为："+a);
            return e;
        }


        private class Node{
            E item;
            Node next;
            Node(E item){
                this.item = item;
            }
        }
    }

}
